package com.asap.demo.sourcefunc;

import com.alibaba.fastjson.JSON;
import com.asap.demo.model.RuleBean;
import com.asap.rule.orm.DbFetcher;
import com.asap.rule.util.RuleReader;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class MysqlSourceFunction2 extends RichSourceFunction<RuleBean> {

	private static final Logger logger = LoggerFactory.getLogger(MysqlSourceFunction.class);

	private PreparedStatement ps;
	private Connection conn;

	@Override
	public void run(SourceContext<RuleBean> ctx) throws Exception {
		ResultSet rs = ps.executeQuery();
		while (rs.next()) {
			String configJson = rs.getString("CONFIG_JSON");
			RuleBean ruleBean = JSON.parseObject(configJson, RuleBean.class);
			logger.info("rule: {}", ruleBean.toString());
			ctx.collect(ruleBean);
		}
	}

	@Override
	public void cancel() {
	}

	@Override
	public void close() throws Exception {
		super.close();
		if (conn != null)
			conn.close();
		if (ps != null)
			ps.close();
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		this.conn = getConn();
		String sql = "select CONFIG_JSON from RE_RULE";
		this.ps = this.conn.prepareStatement(sql);
	}

	@Override
	protected Object clone() throws CloneNotSupportedException {
		return super.clone();
	}

	public static Connection getConn() {
		Connection conn = null;
		try {
			Class.forName("com.mysql.jdbc.Driver");
			conn = DriverManager.getConnection(
					"jdbc:mysql://192.168.1.238:3306/SSA?useUnicode=true&characterEncoding=utf-8",
					"root",
					"1qazXSW@3edc");
		} catch (ClassNotFoundException | SQLException e) {
			e.printStackTrace();
		}
		return conn;
	}

}
